package com.sparrow.socket.tcp.session.impl;

import com.sparrow.common.thread.ThreadPool;
import com.sparrow.socket.tcp.handler.FlowFireHandler;
import com.sparrow.socket.tcp.session.Session;
import io.netty.channel.Channel;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.Timer;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

public class AbstractSession implements Session {

    protected Channel channel;

    protected Object identity;

    protected Object entity;

    private final Map<Object, Object> attributeMap = new ConcurrentHashMap<>();

    private volatile boolean delayFlush = true;

    private FlowFireHandler.FlowFireRecord flowFireRecord;

    public void setFlowFireRecord(FlowFireHandler.FlowFireRecord flowFireRecord) {
        this.flowFireRecord = flowFireRecord;
    }

    public FlowFireHandler.FlowFireRecord getFlowFireRecord() {
        return flowFireRecord;
    }

    private AtomicBoolean writeStatus = new AtomicBoolean(false);

    private volatile Timeout timeout;

    public Timeout getTimeout() {
        return timeout;
    }

    public Channel getChannel() {
        return channel;
    }

    public void setChannel(Channel channel) {
        this.channel = channel;
    }

    @Override
    public Object getIdentity() {
        return identity;
    }

    @Override
    public void setIdentity(Object identity) {
        this.identity = identity;
    }

    @Override
    public String id() {
        return channel == null ? "null" : channel.id().toString();
    }

    @Override
    public String channelInfo() {
        return channel == null ? "null" : channel.toString();
    }

    public Object getEntity() {
        return entity;
    }

    public void setEntity(Object entity) {
        this.entity = entity;
    }

    public Map<Object, Object> getAttributeMap() {
        return attributeMap;
    }

    public Object getAttribute(Object key) {
        return attributeMap.get(key);
    }

    public Object getAttribute(Object key, Object defaultValue) {
        return attributeMap.getOrDefault(key, defaultValue);
    }

    public Object setAttribute(Object key, Object value) {
        return value != null ? attributeMap.put(key, value) : attributeMap.remove(key);
    }

    public void setDelayFlush(boolean delayFlush) {
        this.delayFlush = delayFlush;
    }

    public boolean isDelayFlush() {
        return delayFlush;
    }

    public void write(Object packet) {
        if (!channel.isWritable()) {//netty 高水位暂停写入
            return;
        }
        if (!delayFlush) {
            channel.writeAndFlush(packet);
            return;
        }
        channel.write(packet);
        if (writeStatus.compareAndSet(false, true)) {
            this.timeout = ThreadPool.TIMER.newTimeout(t -> {
                if (writeStatus.compareAndSet(true, false)) {
                    channel.flush();
                }
            }, 50, TimeUnit.MILLISECONDS);
        }
    }

    public void writeAndFlush(Object packet) {
        if (!channel.isActive()) {
            return;
        }
        if (!channel.isWritable()) {
            return;
        }
        channel.writeAndFlush(packet);
    }

    public void flush() {
        channel.flush();
    }

    public void close() {
        if (channel != null) {
            channel.close();
        }
        if (timeout != null) {
            timeout.cancel();
        }
    }

    public boolean isActive() {
        return channel != null && channel.isActive();
    }
}
